package com.rrd.dw.canal.client;

import java.net.InetSocketAddress;
import java.util.List;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.rrd.dw.canal.sink.KafkaSink;

public class PigeonCannlClient extends AbstractCanalClient{

	public PigeonCannlClient(String destination) {
		super(destination);
	}
	public PigeonCannlClient(String destination, CanalConnector connector) {
		super(destination,connector);
	}
	/**
	 * 解析rowdata，然后写入到存储
	 * @param rowData
	 * @param type
	 */
	private void rowDataToString(List<Column> columnList,String opType){
		StringBuilder sb = new StringBuilder(500);
		sb.append(opType);
		for(Column c : columnList){
			sb.append(CRT001).append(c.getValue()==null?"":c.getValue());
		}
		if(this.sink==null){
			printColumn(columnList);
		}else{
			
			this.sink.write(columnList.get(2).getValue(),sb.toString().replace("\n", ""));
		}
		
	}
	public static void main(String args[]) {
        String destination = "pigeon";
        String topic ="cannlTest";
        String zk="172.16.2.161:2181,172.16.2.162:2181";
        logger.info("destination=="+destination);
        logger.info("topic=="+topic);
        logger.info("zookeeper=="+zk);
        // 基于zookeeper动态获取canal server的地址，建立链接，其中一台server发生crash，可以支持failover
        CanalConnector connector = CanalConnectors.newClusterConnector(zk,destination, "", "");
//        String ip = "10.10.15.65";
//        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11112),
//            destination,
//            "",
//            "");
        final PigeonCannlClient client = new PigeonCannlClient(destination);
        final KafkaSink sink = new KafkaSink(topic);
        connector.connect();
		connector.subscribe("pigeon.task");
        client.setConnector(connector);
//        client.setSink(sink);
        client.start();

        Runtime.getRuntime().addShutdownHook(new Thread() {

            public void run() {
                try {
                    logger.info("## stop the canal client");
                    client.stop();
                    sink.closeProducer();
                } catch (Throwable e) {
                    logger.warn("##something goes wrong when stopping canal:", e);
                } finally {
                    logger.info("## canal client is down.");
                }
            }
           
        });
    }

}
